package com.edata.bigdata.util

import org.apache.log4j.Logger
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataType, DataTypes}

import scala.util.matching.Regex

object EDataUtils extends Serializable {

  @transient lazy val LOGGER = Logger.getLogger(this.getClass)

  def convertDataType(dtype: String): DataType = {
    dtype.toLowerCase match {
      case "string" => DataTypes.StringType
      case "int" | "integer" => DataTypes.IntegerType
      case "float" => DataTypes.FloatType
      case "boolean" => DataTypes.BooleanType
      case "double" => DataTypes.DoubleType
      case "date" => DataTypes.DateType
      case "timestamp" => DataTypes.TimestampType
      case "short" => DataTypes.ShortType
      case "point" => DataTypes.StringType
      case "linestring" => DataTypes.StringType
      case x => {
        LOGGER.warn(s"$x data type does't existed")
        DataTypes.NullType
      }
    }
  }

  def convertVToQueryStrByDataType(value: String, dtype: String) = {
    dtype.toLowerCase match {
      case "string" => s"'${value}'"
      case "int" | "integer" => s"${value}"
      case "float" => s"${value}"
      case "short" => s"${value}"
      case "boolean" => s"${value}"
      case "double" => s"${value}"
      case "long" => s"${value}"
      case "date" => s"'${value}'"
      case "timestamp" => s"'${value}'"
    }
  }

  def convertKToQueryStrByDataType(key: String, dtype: String): String = {
    dtype.toLowerCase match {
      case "string" => s"${key}"
      case "int" | "integer" => s"${key}"
      case "float" => s"${key}"
      case "short" => s"${key}"
      case "boolean" => s"${key}"
      case "double" => s"${key}"
      case "long" => s"${key}"
      case "date" => s"${key}"
      case "timestamp" => s"${key}"
      case "point" => s"ST_asText(${key}) as ${key}"
      case "linestring" => s"ST_asText(${key}) as ${key}"
    }
  }

  def convertVToSaveStrByDataType(value: String, dtype: String): String = {
    dtype.toLowerCase match {
      case "string" => s"'${value}'"
      case "int" | "integer" => s"${value}"
      case "float" => s"${value}"
      case "short" => s"${value}"
      case "boolean" => s"${value}"
      case "double" => s"${value}"
      case "long" => s"${value}"
      case "date" => s"'${value}'"
      case "timestamp" => s"'${value}'"
      case "point" => s"ST_GeogFromText('${value}')"
      case "linestring" => s"ST_GeogFromText('${value}')"

    }
  }

  def findColValOfRowByColName(row: Row, colName: String, dtype: String): String = {
    dtype.toLowerCase match {
      case "string" => row.getAs[String](colName)
      case "int" | "integer" => row.getAs[Int](colName).toString
      case "float" => row.getAs[Float](colName).toString
      case "double" => row.getAs[Double](colName).toString
      case "boolean" => row.getAs[Boolean](colName).toString
      case "short" => row.getAs[Boolean](colName).toString
      case "long" => row.getAs[Long](colName).toString
      case "date" => row.getAs[String](colName)
      case "timestamp" => row.getAs[String](colName)
      case "point" =>{
        val data = findStrByRegex(row.getAs[String](colName),"POINT\\(([^\\)]+)\\)")
        s"POINT(${data})"
      }
      case "linestring" => {
        val data = findStrByRegex(row.getAs[String](colName),"LINESTRING\\(\\(([^\\)]+)\\)\\)")
        s"LINESTRING(${data})"
      }
      case x => row.getAs[String](colName)
    }
  }

  def findStrByRegex(str:String,regex:String):String = {
    val pattern = new Regex(regex)
    val matched = pattern.findFirstMatchIn(str)
    var result = ""
    matched.foreach{m=>
      result = m.group(1)
    }
    result
  }



}